/*
 * Copyright (c) 2009-2011, Salvatore Sanfilippo <antirez at gmail dot com>
 * Copyright (c) 2010-2011, Pieter Noordhuis <pcnoordhuis at gmail dot com>
 *
 * All rights reserved.
 *
 * Redistribution and use in source and binary forms, with or without
 * modification, are permitted provided that the following conditions are met:
 *
 *   * Redistributions of source code must retain the above copyright notice,
 *     this list of conditions and the following disclaimer.
 *   * Redistributions in binary form must reproduce the above copyright
 *     notice, this list of conditions and the following disclaimer in the
 *     documentation and/or other materials provided with the distribution.
 *   * Neither the name of Redis nor the names of its contributors may be used
 *     to endorse or promote products derived from this software without
 *     specific prior written permission.
 *
 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
 * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
 * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
 * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE
 * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
 * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
 * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
 * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
 * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
 * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
 * POSSIBILITY OF SUCH DAMAGE.
 */

#include <stdlib.h>
#include <string.h>
#ifndef _WIN32
#include <strings.h>
#endif
#include <assert.h>
#include <ctype.h>
#include <errno.h>
#include "async.h"
#include "dict.c"
#include "sds.h"
#include "ELoggingHeader.h"

#ifdef _WIN32
#define strcasecmp _stricmp
#define strncasecmp _strnicmp
#endif

#include <map>
#include <string>

using namespace std;

/* Functions that proxy to hiredis */

/* Command functions for an async context. Write the command to the
 * output buffer and register the provided callback. */
struct redisCallback {
	redisCallback(redisCallbackFn fun):fn(fun){}
    redisCallback *next; /* simple singly linked list */
    redisCallbackFn fn;
    void *privdata;
	redisCallback():next(NULL), privdata(NULL){}
};

/* List of callbacks for either regular replies or pub/sub */
struct redisCallbackList {
    redisCallback *head, *tail;
};

/* Context for an async connection to Redis */
struct  redisAsyncContext {
	/* Hold the regular context, so it can be     ealloc'ed. */
	redisAsyncContext(SendDataCallbackFn sSendDataCallbackFn, DisconnectFn sDisconnectFn);
    redisContext predisContext;

    /* Setup error flags so they can be used directly. */
    int err;
    char *errstr;

	SendDataCallbackFn		pSendDataCallbackFn;
    /* Called when either the connection is terminated due to an error or per
     * user request. The status is set accordingly (REDIS_OK, REDIS_ERR). */
    DisconnectFn		onDisconnect;

    /* Called when the first write event was received. */
    redisSessionCallback onConnect;

    /* Regular command callbacks */
    redisCallbackList replies;

    /* Subscription callbacks */
    struct {
        redisCallbackList invalid;
		map<string, redisCallback> channels;
		map<string, redisCallback> patterns;
    } sub;
};

/* Functions managing dictionary of callbacks for pub/sub. */
static unsigned int callbackHash(const void *key) {
    return dictGenHashFunction((const unsigned char *)key,
                               (int)sdslen((const sds)key));
}

static void *callbackValDup(void *privdata, const void *src) {
    redisCallback *dup = (redisCallback*)malloc(sizeof(*dup));
    ((void) privdata);
    memcpy(dup,src,sizeof(*dup));
    return dup;
}

static int callbackKeyCompare(void *privdata, const void *key1, const void *key2) {
    size_t l1, l2;
    ((void) privdata);

    l1 = sdslen((const sds)key1);
    l2 = sdslen((const sds)key2);
    if (l1 != l2) return 0;
    return memcmp(key1,key2,l1) == 0;
}

static void callbackKeyDestructor(void *privdata, void *key) {
    ((void) privdata);
    sdsfree((sds)key);
}

static void callbackValDestructor(void *privdata, void *val) {
    ((void) privdata);
    free(val);
}

static dictType callbackDict = {
    callbackHash,
    NULL,
    callbackValDup,
    callbackKeyCompare,
    callbackKeyDestructor,
    callbackValDestructor
};

/* We want the error field to be accessible directly instead of requiring
 * an indirection to the redisContext struct. */
static void __redisAsyncCopyError(redisAsyncContext *ac) {
    redisContext *c = &(ac->predisContext);
    ac->err = c->err;
    ac->errstr = c->errstr;
}

/* Helper functions to push/shift callbacks */
static int __redisPushCallback(redisCallbackList *list, redisCallback *source) {
    redisCallback *cb = NULL;
	if (source){
		cb = new redisCallback(source->fn);
		cb->next = NULL;
		cb->privdata = source->privdata;
	}
	else{
		cb = new redisCallback(NULL);
	}

    /* Copy callback from stack to heap */
    if (cb == NULL)
        return REDIS_ERR_OOM;

    /* Store callback in list */
    if (list->head == NULL)
        list->head = cb;
    if (list->tail != NULL)
        list->tail->next = cb;
    list->tail = cb;
    return REDIS_OK;
}

static int __redisShiftCallback(redisCallbackList *list, redisCallback *& target) {
    redisCallback *cb = list->head;
    if (cb != NULL) {
        list->head = cb->next;
        if (cb == list->tail)
            list->tail = NULL;

		target = cb;
        return REDIS_OK;
    }
    return REDIS_ERR;
}

static void __redisRunCallback(redisAsyncContext *ac, redisCallback *cb, redisReply *reply) {
    redisContext *c = &(ac->predisContext);
    if (cb->fn != NULL) {
        c->flags |= REDIS_IN_CALLBACK;
        cb->fn(ac,reply,cb->privdata);
        c->flags &= ~REDIS_IN_CALLBACK;
    }
}

/* Helper function to free the context. */
static void __redisAsyncFree(redisAsyncContext *ac) {
    //redisContext *c = &(ac->predisContext);
    //redisCallback* cb = NULL;
    //dictIterator *it;
    //dictEntry *de;

    ///* Execute pending callbacks with NULL reply. */
    //while (__redisShiftCallback(&ac->replies, cb) == REDIS_OK)
    //    __redisRunCallback(ac, cb,NULL);

    ///* Execute callbacks for invalid commands */
    //while (__redisShiftCallback(&ac->sub.invalid,cb) == REDIS_OK)
    //    __redisRunCallback(ac, cb,NULL);

    ///* Run subscription callbacks callbacks with NULL reply */
    //it = dictGetIterator(ac->sub.channels);
    //while ((de = dictNext(it)) != NULL)
    //    __redisRunCallback(ac,(redisCallback*)dictGetEntryVal(de),NULL);
    //dictReleaseIterator(it);
    //dictRelease(ac->sub.channels);

    //it = dictGetIterator(ac->sub.patterns);
    //while ((de = dictNext(it)) != NULL)
    //    __redisRunCallback(ac,(redisCallback*)dictGetEntryVal(de),NULL);
    //dictReleaseIterator(it);
    //dictRelease(ac->sub.patterns);

    ///* Signal event lib to clean up */
    ///* Execute disconnect callback. When redisAsyncFree() initiated destroying
    // * this context, the status will always be REDIS_OK. */
    //if (ac->onDisconnect && (c->flags & REDIS_CONNECTED)) {
    //    if (c->flags & REDIS_FREEING) {
    //        ac->onDisconnect(REDIS_OK);
    //    } else {
    //        ac->onDisconnect((ac->err == 0) ? REDIS_OK : REDIS_ERR);
    //    }
    //}

    ///* Cleanup self */
    //redisFree(c);
}

/* Free the async context. When this function is called from a callback,
 * control needs to be returned to redisProcessCallbacks() before actual
 * free'ing. To do so, a flag is set on the context which is picked up by
 * redisProcessCallbacks(). Otherwise, the context is immediately free'd. */
void redisAsyncContextWrapper::redisAsyncFree(redisAsyncContext *ac) {
    redisContext *c = &(ac->predisContext);
    c->flags |= REDIS_FREEING;
    if (!(c->flags & REDIS_IN_CALLBACK))
        __redisAsyncFree(ac);
}

/* Helper function to make the disconnect happen and clean up. */
static void __redisAsyncDisconnect(redisAsyncContext *ac) {
    redisContext *c = &(ac->predisContext);

    /* Make sure error is accessible if there is any */
    __redisAsyncCopyError(ac);

	redisCallback* cb = NULL;

    if (ac->err == 0) {
        /* For clean disconnects, there should be no pending callbacks. */
        assert(__redisShiftCallback(&ac->replies, cb) == REDIS_ERR);
    } else {
        /* Disconnection is caused by an error, make sure that pending
         * callbacks cannot call new commands. */
        c->flags |= REDIS_DISCONNECTING;
    }

    /* For non-clean disconnects, __redisAsyncFree() will execute pending
     * callbacks with a NULL-reply. */
    __redisAsyncFree(ac);
}

static int __redisGetSubscribeCallback(redisAsyncContext *ac, redisReply *reply, redisCallback *& dstcb) {
    redisContext *c = &(ac->predisContext);
    std::map<string, redisCallback> *callbacks;
    int pvariant;
    char *stype;

    /* Custom reply functions are not supported for pub/sub. This will fail
     * very hard when they are used... */
    if (reply->type == REDIS_REPLY_ARRAY) {
        assert(reply->elements >= 2);
        assert(reply->element[0]->type == REDIS_REPLY_STRING);
        stype = reply->element[0]->str;
        pvariant = (tolower(stype[0]) == 'p') ? 1 : 0;

        if (pvariant)
            callbacks = &ac->sub.patterns;
        else
            callbacks = &ac->sub.channels;

        /* Locate the right callback */
        assert(reply->element[1]->type == REDIS_REPLY_STRING);
        string sname(reply->element[1]->str,reply->element[1]->len);
		auto iter = callbacks->find(sname);
		if (iter != callbacks->end()){
			dstcb = &iter->second;

            /* If this is an unsubscribe message, remove it. */
            if (strcasecmp(stype+pvariant,"unsubscribe") == 0) {
				callbacks->erase(iter);
                /* If this was the last unsubscribe message, revert to
                 * non-subscribe mode. */
                assert(reply->element[2]->type == REDIS_REPLY_INTEGER);
                if (reply->element[2]->integer == 0)
                    c->flags &= ~REDIS_SUBSCRIBED;
            }
        }
    } else {
        /* Shift callback for invalid commands. */
        __redisShiftCallback(&ac->sub.invalid,dstcb);
    }
    return REDIS_OK;
}

/* Sets a pointer to the first argument and its length starting at p. Returns
 * the number of bytes to skip to get to the following argument. */
static char *nextArgument(char *start, char **str, size_t *len) {
    char *p = start;
    if (p[0] != '$') {
        p = strchr(p,'$');
        if (p == NULL) return NULL;
    }

    *len = (int)strtol(p+1,NULL,10);
    p = strchr(p,'\r');
    assert(p);
    *str = p+2;
    return p+2+(*len)+2;
}

/* Helper function for the redisAsyncCommand* family of functions. Writes a
 * formatted command to the output buffer and registers the provided callback
 * function with the context. */
static int __redisAsyncCommand(redisAsyncContext *ac, redisCallbackFn fn, void *privdata, char *cmd, size_t len) {
    redisContext *c = &(ac->predisContext);
    redisCallback cb(fn);
    int pvariant, hasnext;
    char *cstr, *astr;
    size_t clen, alen;
    char *p;

    /* Don't accept new commands when the connection is about to be closed. */
    if (c->flags & (REDIS_DISCONNECTING | REDIS_FREEING)) return REDIS_ERR;

    /* Setup callback */
    cb.privdata = privdata;

    /* Find out which command will be appended. */
    p = nextArgument(cmd,&cstr,&clen);
    assert(p != NULL);
    hasnext = (p[0] == '$');
    pvariant = (tolower(cstr[0]) == 'p') ? 1 : 0;
    cstr += pvariant;
    clen -= pvariant;

    if (hasnext && strncasecmp(cstr,"subscribe\r\n",11) == 0) {
        c->flags |= REDIS_SUBSCRIBED;

        /* Add every channel/pattern to the list of subscription callbacks. */
        while ((p = nextArgument(p,&astr,&alen)) != NULL) {
            string sname(astr,alen);
            if (pvariant)
				ac->sub.patterns[sname] = cb;
            else
				ac->sub.channels[sname] = cb;
        }
    } else if (strncasecmp(cstr,"unsubscribe\r\n",13) == 0) {
        /* It is only useful to call (P)UNSUBSCRIBE when the context is
         * subscribed to one or more channels or patterns. */
        if (!(c->flags & REDIS_SUBSCRIBED)) return REDIS_ERR;

        /* (P)UNSUBSCRIBE does not have its own response: every channel or
         * pattern that is unsubscribed will receive a message. This means we
         * should not append a callback function for this command. */
     } else if(strncasecmp(cstr,"monitor\r\n",9) == 0) {
         /* Set monitor flag and push callback */
         c->flags |= REDIS_MONITORING;
         __redisPushCallback(&ac->replies,&cb);
    } else {
        if (c->flags & REDIS_SUBSCRIBED)
            /* This will likely result in an error reply, but it needs to be
             * received and passed to the callback. */
            __redisPushCallback(&ac->sub.invalid,&cb);
        else
            __redisPushCallback(&ac->replies,&cb);
    }

	ac->pSendDataCallbackFn(cmd, len);

    return REDIS_OK;
}

int redisAsyncContextWrapper::redisAsyncCommand(redisCallbackFn fn, void *privdata, const char *format, ...){
	if (!m_IsConnected){
		return REDIS_DISCONNECTING;
	}
	va_list ap;
	int status;
	va_start(ap,format);
	status = AsyncCommand(fn,privdata,format,ap);
	va_end(ap);
	return status;
}

int redisAsyncContextWrapper::AsyncCommand(redisCallbackFn fn, void *privdata, const char *format, va_list ap){
	char *cmd;
	int len;
	int status;
	len = redisvFormatCommand(&cmd,format,ap);
	status = __redisAsyncCommand(m_predisAsyncContext,fn,privdata,cmd,len);
	free(cmd);
	return status;
}

void redisAsyncContextWrapper::redisProcessCallbacks(){
	redisAsyncContext* ac = m_predisAsyncContext;
	redisContext *c = &(ac->predisContext);
    redisCallback* cb = NULL;
    void *reply = NULL;
    int status;

    while((status = redisGetReply(c,&reply)) == REDIS_OK) {
        if (reply == NULL) {
            /* When the connection is being disconnected and there are
             * no more replies, this is the cue to really disconnect. */
            if (c->flags & REDIS_DISCONNECTING && sdslen(c->obuf) == 0) {
                __redisAsyncDisconnect(ac);
                return;
            }
            break;
        }

		bool ifSubscribeCallback = c->flags & REDIS_SUBSCRIBED;
        /* Even if the context is subscribed, pending regular callbacks will
         * get a reply before pub/sub messages arrive. */
        if (__redisShiftCallback(&ac->replies, cb) != REDIS_OK) {
            /*
             * A spontaneous reply in a not-subscribed context can be the error
             * reply that is sent when a new connection exceeds the maximum
             * number of allowed connections on the server side.
             *
             * This is seen as an error instead of a regular reply because the
             * server closes the connection after sending it.
             *
             * To prevent the error from being overwritten by an EOF error the
             * connection is closed here. See issue #43.
             *
             * Another possibility is that the server is loading its dataset.
             * In this case we also want to close the connection, and have the
             * user wait until the server is ready to take our request.
             */
            if (((redisReply*)reply)->type == REDIS_REPLY_ERROR) {
                c->err = REDIS_ERR_OTHER;
                _snprintf(c->errstr,sizeof(c->errstr),"%s",((redisReply*)reply)->str);
                __redisAsyncDisconnect(ac);
				
				if (cb){
					delete cb;
					cb = nullptr;
				}

                return;
            }
            /* No more regular callbacks and no errors, the context *must* be subscribed or monitoring. */
            assert((c->flags & REDIS_SUBSCRIBED || c->flags & REDIS_MONITORING));
            if(ifSubscribeCallback)
                __redisGetSubscribeCallback(ac,(redisReply*)reply, cb);
        }

		if (cb == nullptr){
			continue;
		}

        if (cb->fn != NULL) {
            __redisRunCallback(ac, cb,(redisReply*)reply);
            c->reader->fn->freeObject(reply);

            /* Proceed with free'ing when redisAsyncFree() was called. */
            if (c->flags & REDIS_FREEING) {
                __redisAsyncFree(ac);

				if(!ifSubscribeCallback)
				{
					delete cb;
					cb = nullptr;
				}

                return;
            }
        } else {
            /* No callback for this reply. This can either be a NULL callback,
             * or there were no callbacks to begin with. Either way, don't
             * abort with an error, but simply ignore it because the client
             * doesn't know what the server will spit out over the wire. */
			redisReply* pReply = (redisReply*)reply;
			if (pReply->type == REDIS_REPLY_ERROR){
				ELOG(LOG_ERROR, "RedisError for %s", pReply->str);
			}
			
            c->reader->fn->freeObject(reply);
        }

		if (!ifSubscribeCallback){
			delete cb;
			cb = nullptr;
		}
    }
}

int redisAsyncContextWrapper::redisAsyncCommandArgv(redisCallbackFn fn, void *privdata, int argc, const char **argv, const size_t *argvlen){
	char *cmd;
	int len;
	int status;
	len = redisFormatCommandArgv(&cmd,argc,argv,argvlen);
	status = __redisAsyncCommand(m_predisAsyncContext,fn,privdata,cmd,len);
	free(cmd);
	return status;
}

int redisAsyncContextWrapper::redisAsyncReaderFeed(const char *buf, size_t len){
	return redisReaderFeed(m_predisAsyncContext->predisContext.reader, buf, len);
}

void redisAsyncContextWrapper::SetSendCallBack(SendDataCallbackFn sFun){
	m_predisAsyncContext->pSendDataCallbackFn = sFun;
}

redisAsyncContextWrapper::redisAsyncContextWrapper():m_IsConnected(false){

}

void redisAsyncContextWrapper::Init(SendDataCallbackFn sFun, DisconnectFn sDisconnectFn){
	m_predisAsyncContext = new redisAsyncContext(sFun, sDisconnectFn);
}

int redisAsyncContextWrapper::redisvAsyncWithBinaryCommand(redisCallbackFn fn, void *privdata, const char *data, size_t length){
	redisContext *c = &(m_predisAsyncContext->predisContext);
    redisCallback cb(fn);
    int pvariant, hasnext;
    char *cstr, *astr;
    size_t clen, alen;
    char *p;
    sds sname;

    /* Don't accept new commands when the connection is about to be closed. */
    if (c->flags & (REDIS_DISCONNECTING | REDIS_FREEING)) return REDIS_ERR;

    /* Setup callback */
    cb.privdata = privdata;
    __redisPushCallback(&m_predisAsyncContext->replies,&cb);

	m_predisAsyncContext->pSendDataCallbackFn(data, length);

    return REDIS_OK;
}

bool redisAsyncContextWrapper::CanbeUsed(){
	return m_IsConnected;
}

void redisAsyncContextWrapper::SetConnected(bool flag){
	m_IsConnected = flag;
}

redisAsyncContext::redisAsyncContext(SendDataCallbackFn sSendDataCallbackFn, DisconnectFn sDisconnectFn):pSendDataCallbackFn(sSendDataCallbackFn)
, onConnect(NULL)
,	onDisconnect(sDisconnectFn){
	    /* The regular connect functions will always set the flag REDIS_CONNECTED.
     * For the async API, we want to wait until the first write event is
     * received up before setting this flag, so reset it here. */
   predisContext.flags = REDIS_CONNECTED;
   err = 0;
   errstr = NULL;
   replies.head = NULL;
   replies.tail = NULL;
   sub.invalid.head = NULL;
   sub.invalid.tail = NULL;
}